package com.web;

import info.monitorenter.cpdetector.io.*;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.*;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

import static java.lang.Boolean.FALSE;
import static java.lang.Boolean.TRUE;

/**
 * @author zhang guoxiang
 * @date 2023/7/24
 */

public class FileSourceFunction implements SourceFunction<String> {

    private String logPath;
    private Long index;

    public FileSourceFunction(String logPath, Long index) {
        this.logPath = logPath;
        this.index = index;
    }


    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        String charset = getCharsetName(new File(logPath));
        index = loadIndex(logPath);
        for (; ; ) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(logPath, "r")) {
                randomAccessFile.seek(index);
                String line = null;
                while ((line = randomAccessFile.readLine()) != null) {
                    if (line.length() > 0) {
                        String data = new String(line.getBytes(StandardCharsets.ISO_8859_1), charset);
                        ctx.collect(data);
                        index = randomAccessFile.getFilePointer() - 1;
                        saveIndex(logPath, index);
                    }
                }
            }
            TimeUnit.MILLISECONDS.sleep(500);
        }
    }


    /**
     * 获取文件编码字符串
     *
     * @param file
     * @return
     */
    public static String getCharsetName(File file) throws IOException {
        String charsetName = "UTF-8";
        // 获取 CodepageDetectorProxy 实例
        CodepageDetectorProxy detector = CodepageDetectorProxy.getInstance();
        // 添加解析器，会使用到添加的后 2 个 ext 里的 jar 包
        detector.add(new ParsingDetector(false));
        detector.add(JChardetFacade.getInstance());
        detector.add(ASCIIDetector.getInstance());
        detector.add(UnicodeDetector.getInstance());
        Charset charset = detector.detectCodepage(file.toURI().toURL());
        if (charset != null) {
            charsetName = charset.name();
        }
        return charsetName;
    }


    @Override
    public void cancel() {

    }


    private Long loadIndex(String path) throws IOException {
        path = DigestUtils.md5Hex(path);
        Properties properties = new Properties();
        File file = new File("conf/index.properties");
        if (FALSE.equals(file.exists())) {
            file.createNewFile();
        }
        try (InputStream inputStream = Files.newInputStream(file.toPath())) {
            properties.load(inputStream);
        }
        return Objects.isNull(properties.get(path)) ? 0L : Long.parseLong(properties.get(path).toString());
    }

    private void saveIndex(String path, Long index) throws IOException {
        Properties properties = new Properties();
        path = DigestUtils.md5Hex(path);
        File file = new File("conf/index.properties");
        if (FALSE.equals(file.exists())) {
            file.createNewFile();
        }
        try (InputStream inputStream = Files.newInputStream(file.toPath()); FileOutputStream fileOutputStream = new FileOutputStream(file)) {
            properties.load(inputStream);
            properties.setProperty(path, index.toString());
            properties.store(fileOutputStream, "");
        }
    }
}
